package de.distMLP.evaluation;

import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.text.DecimalFormat;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hama.bsp.BSP;
import org.apache.hama.bsp.BSPPeer;
import org.apache.hama.bsp.sync.SyncException;

import de.distMLP.messages.HamaMessage;
import de.distMLP.train.Training_Helper;
import de.mlp_distributed.mlp.core.MultiLayerPerceptron;
import de.mlp_distributed.mlp.math.Factory;
import de.mlp_distributed.mlp.math.mahout.Matrix;
import de.mlp_distributed.mlp.math.mahout.Vector;

public class MLPEvaluator extends BSP<LongWritable, Text, NullWritable, Text, HamaMessage> {

	private MultiLayerPerceptron mlp;
	private static final DecimalFormat df = new DecimalFormat("#.#####");

	private enum Confusion {
		TP, FP, FN, TN;
	}

	private class EvaluationResult {
		public EvaluationResult(final Confusion confusion, final double target, final double result) {
			this.confusion = confusion;
			this.result = result;
			this.target = target;
		}

		Confusion confusion;
		double result, target;
	}

	private Map<Integer, String> readDictionary(final String filenamePath) throws IOException {
		BufferedReader br = null;
		final Map<Integer, String> dictionary = new HashMap<Integer, String>();
		try {

			String sCurrentLine;

			br = new BufferedReader(new FileReader(filenamePath));

			while ((sCurrentLine = br.readLine()) != null) {
				final String[] split = sCurrentLine.split(":");
				if (split.length == 2) {
					dictionary.put(Integer.parseInt(split[0]), split[1]);
				}
			}

		} catch (final IOException e) {
			e.printStackTrace();
		} finally {
			try {
				if (br != null) {
					br.close();
				}
			} catch (final IOException ex) {
				ex.printStackTrace();
			}
		}
		return dictionary;
	}

	private double computeStandardDeviation(final List<EvaluationResult> resultList, final Confusion cf) {
		if (resultList.size() == 0) {
			return 0;
		}
		final double target = resultList.get(0).target;

		double average = 0;
		int size = 0;
		for (int i = 0; i < resultList.size(); i++) {
			final EvaluationResult er = resultList.get(i);
			if (er.confusion == cf) {
				size++;
				average += er.result;
			}
		}
		average = average / size;

		double sum = 0;
		for (int j = 0; j < resultList.size(); j++) {
			final EvaluationResult er = resultList.get(j);
			if (er.confusion == cf) {
				sum += Math.pow(resultList.get(j).result - average, 2);
			}
		}
		sum = (sum) / (size - 1);
		final double stdDev = Math.sqrt(sum);

		final double unsecurity = this.computeUnsecurity(stdDev, size);

		System.out.println("class: " + target + " confusion: (" + size + ") " + cf + " average: " + average + " stdDev: " + stdDev
				+ " unsecurity: " + unsecurity);

		return stdDev;
	}

	private double computeUnsecurity(final double standardDeviation, final int n) {
		final double unsecurity = standardDeviation / Math.sqrt(n);
		return unsecurity;
	}

	@Override
	public void bsp(final BSPPeer<LongWritable, Text, NullWritable, Text, HamaMessage> peer) throws IOException, SyncException,
			InterruptedException {
		final String dictPath = peer.getConfiguration().get(de.distMLP.train.Configuration.DICTIONARY_PATH);
		Map<Integer, String> dictionary = null;
		if ((dictPath != null) && !dictPath.isEmpty()) {
			dictionary = this.readDictionary(dictPath);
		}

		final LongWritable key = new LongWritable();
		final Text value = new Text();

		int tp = 0, fn = 0, fp = 0, tn = 0;

		final List<EvaluationResult> classZeroList = new ArrayList<MLPEvaluator.EvaluationResult>();
		final List<EvaluationResult> classOneList = new ArrayList<MLPEvaluator.EvaluationResult>();

		while (peer.readNext(key, value)) {
			if (this.mlp.getNumberOfOutputUnits() == 1) {
				final EvaluationResult er = this.evaluateOneClass(peer, value);

				final Confusion confusion = er.confusion;
				if (confusion == null) {
					continue;
				}

				if (er.target == 0.0) {
					classZeroList.add(er);
				} else {
					classOneList.add(er);
				}

				switch (confusion) {
				case TP:
					tp++;
					break;
				case FP:
					fp++;
					break;
				case FN:
					fn++;
					break;
				case TN:
					tn++;
					break;
				}
			} else {
				this.evaluate(peer, value, dictionary);
			}
		}

		if (this.mlp.getNumberOfOutputUnits() == 1) {
			if (((tp + fn) > 0) && ((tp + fp) > 0)) {
				final double recall = (double) tp / (double) (tp + fn);
				final double precision = ((double) tp / (double) (tp + fp));
				final double f1measure = (2 * precision * recall) / (precision + recall);
				peer.write(NullWritable.get(), new Text("Precision: " + precision + " Recall: " + recall + " F1-measure " + f1measure));

				this.computeStandardDeviation(classOneList, Confusion.TP);
				this.computeStandardDeviation(classOneList, Confusion.FN);

				this.computeStandardDeviation(classZeroList, Confusion.FP);
				this.computeStandardDeviation(classZeroList, Confusion.TN);
			} else {
				peer.write(NullWritable.get(), new Text("Could not calculate P/R! Division by zero!"));
			}
		}

		if (this.mlp.getNumberOfOutputUnits() > 1) {
			this.computeMaxInputForHiddenUnits(peer, dictionary);
		}
	}

	void evaluate(final BSPPeer<LongWritable, Text, NullWritable, Text, HamaMessage> peer, final Text value,
			final Map<Integer, String> dictionary) throws IOException {
		final Vector featureVector = Training_Helper.getFeatureVectorFromTrainingsdata(value, this.mlp.getNumberOfInputUnits());
		final Vector result = this.mlp.classifyFull(featureVector);

		final List<ClassificationResult> sortedResultList = Helper.createList(result);
		final String sb = this.createResultString(dictionary, sortedResultList);
		peer.write(NullWritable.get(), new Text(sb.toString()));
	}

	public String createResultString(final Map<Integer, String> dictionary, final List<ClassificationResult> sortedResultList) {
		int counter = 0;
		final StringBuilder sb = new StringBuilder();
		for (int i = sortedResultList.size() - 1; i >= 0; i--) {
			final ClassificationResult classificationResult = sortedResultList.get(i);
			// Print first 15 items
			if (counter == 15) {
				break;
			}
			if (dictionary != null) {
				// String result =
				// dictionary.get(classificationResult.getNbOutputNeuron()) +
				// " ("
				// +
				// MLPEvaluator.df.format(classificationResult.getOutputValue())
				// + "), ";
				final String result = dictionary.get(classificationResult.getNbOutputNeuron()) + ", ";
				sb.append(result);
			}

			counter++;
		}
		if (sb.toString().length() > 2) {
			return sb.substring(0, sb.length() - 2);
		}
		return "";
	}

	void computeMaxInputForHiddenUnits(final BSPPeer<LongWritable, Text, NullWritable, Text, HamaMessage> peer,
			final Map<Integer, String> dictionary) throws IOException {
		final Matrix[] weightMatrix = this.mlp.getWeights();
		final Matrix layerOne = weightMatrix[0];
		// skip bias
		// j equals the number of input units without bias
		// i equals the number of hidden units in layer 1 without bias

		peer.write(NullWritable.get(), new Text("######## paradigmatic relations ##########"));

		for (int i = 1; i < layerOne.rowSize(); i++) {
			final Vector result = Factory.getInstance().getVectorFactory().construct(layerOne.columnSize());
			final double sum = this.calcSquaredSum(layerOne, i);
			for (int j = 0; j < layerOne.columnSize(); j++) {
				if (sum > 0) {
					result.setQuick(j, layerOne.getQuick(i, j) / sum);
				} else {
					result.setQuick(j, 0);
				}
			}
			final List<ClassificationResult> sortedResultList = Helper.createList(result);
			final String sb = this.createResultString(dictionary, sortedResultList);
			peer.write(NullWritable.get(), new Text(sb));
		}
	}

	double calcSquaredSum(final Matrix layerOne, final int i) {
		double sum = 0;

		for (int j = 0; j < layerOne.columnSize(); j++) {
			sum += Math.pow(layerOne.getQuick(i, j), 2);
		}

		sum = Math.sqrt(sum);

		return sum;
	}

	EvaluationResult evaluateOneClass(final BSPPeer<LongWritable, Text, NullWritable, Text, HamaMessage> peer, final Text value)
			throws IOException {
		final Vector targetVector = Training_Helper.getTargetVectorFromTrainingsdata(value, this.mlp.getNumberOfOutputUnits());
		final Vector result = this.mlp.classifyFull(Training_Helper.getFeatureVectorFromTrainingsdata(value,
				this.mlp.getNumberOfInputUnits()));

		for (final de.mlp_distributed.mlp.math.mahout.Vector.Element outputNeuron : result) {
			// peer.write(NullWritable.get(), new Text("result: " +
			// outputNeuron.get() + " expected: " + targetVector.get(0)));
		}

		final Confusion confusion = this.calculateConfusion(targetVector, result);
		final EvaluationResult er = new EvaluationResult(confusion, targetVector.getQuick(0), result.getQuick(0));
		return er;
	}

	public Confusion calculateConfusion(final Vector targetVector, final Vector result) {
		if (targetVector.size() != 1) {
			return null;
		}

		final double output = result.getQuick(0);
		final double target = targetVector.getQuick(0);

		if (output >= 0.5) {
			if (target == 1) {
				return Confusion.TP;
			} else {
				return Confusion.FP;
			}
		} else {
			if (target == 1) {
				return Confusion.FN;
			} else {
				return Confusion.TN;
			}
		}
	}

	@Override
	public void setup(final BSPPeer<LongWritable, Text, NullWritable, Text, HamaMessage> peer) throws IOException, SyncException,
			InterruptedException {
		super.setup(peer);
		final Configuration conf = peer.getConfiguration();
		final String mlpInputPath = conf.get(de.distMLP.train.Configuration.MLP_INPUT_PATH);
		this.mlp = Training_Helper.readNetwork(peer.getConfiguration(), new Path(mlpInputPath));
	}

}
